#include "empty_function.h"

Mqtt_Token_Response mqtt_token_response;
Mqtt_Connect_Info mqtt_connect_info = {{0}, NULL, false};

static pthread_mutex_t mqtt_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t mqtt_cond   = PTHREAD_COND_INITIALIZER;

//调用shell命令并获取返回值
int shell_cmd(char *cmd, char *buff, int size)
{
    char temp[TOKEN_RESP_BUF_MAX_LEN];
    FILE *fp   = NULL;
    int offset = 0;
    int len    = 0;

    fp = popen(cmd, "r");
    if(fp == NULL)
    {
        return -1;
    }

    while(fgets(temp, sizeof(temp), fp) != NULL)
    {
        len = strlen(temp);
        if(offset + len < size)
        {
            strcpy(buff + offset, temp);
            offset += len;
        }
        else
        {
            buff[offset] = 0;
            break;
        }
    }

    if(fp != NULL)
    {
        pclose(fp);
    }

    return strlen(buff);
}

//通过shell命令获取token,并使用cjson库进行解析
static int get_token_from_shell(Mqtt_Token_Response *mqtt_token_response)
{
    //char *pret                              = NULL;
    int iret                                = -1;
    char token_resp[TOKEN_RESP_BUF_MAX_LEN] = {0};
    iret                                    = shell_cmd(GET_MQTT_TOKEN_SHELL, token_resp, TOKEN_RESP_BUF_MAX_LEN);
    if(iret > 0)
    {
        /*
        cJSON *json = cJSON_Parse(token_resp);
        if(NULL == json)
        {
            return -1;
        }
        cJSON *j_msg = cJSON_GetObjectItem(json, "msg");
        if(NULL == j_msg)
        {
            iret = -2;
            goto PARSE_JSON_EXIT;
        }
        printf("msg:%s\n", j_msg->valuestring);

        //expires_in
        cJSON *j_expires = cJSON_GetObjectItem(json, "expires");
        if(NULL == j_expires)
        {
            iret = -3;
            goto PARSE_JSON_EXIT;
        }
        mqtt_token_response->expires = j_expires->valueint;

        cJSON *j_code = cJSON_GetObjectItem(json, "code");
        if(NULL == j_code)
        {
            iret = -4;
            goto PARSE_JSON_EXIT;
        }
        printf("code:%d\n", j_code->valueint);
        if(j_code->valueint != 1)
        {
            return j_code->valueint;
        }

        //access_token
        cJSON *j_token = cJSON_GetObjectItem(json, "token");
        if(NULL == j_token)
        {
            iret = -5;
            goto PARSE_JSON_EXIT;
        }
        int token_len              = strlen(j_token->valuestring) + 1;
        mqtt_token_response->token = malloc(token_len);
        if(mqtt_token_response->token == NULL)
        {
            iret = -6;
            goto PARSE_JSON_EXIT;
        }
        memcpy(mqtt_token_response->token, j_token->valuestring, token_len);

        //clientId
        cJSON *j_clientId = cJSON_GetObjectItem(json, "clientId");
        if(NULL == j_clientId)
        {
            iret = -7;
            goto PARSE_JSON_EXIT;
        }
        int clientId_len              = strlen(j_clientId->valuestring) + 1;
        mqtt_token_response->clientId = malloc(clientId_len);
        if(mqtt_token_response->clientId == NULL)
        {
            iret = -8;
            goto PARSE_JSON_EXIT;
        }
        memcpy(mqtt_token_response->clientId, j_clientId->valuestring, clientId_len);

        iret = 0;
    PARSE_JSON_EXIT:
        cJSON_Delete(json);
        return iret;

        */
    }
    return -100;
}

static void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
    printf("Reconnecting\n");

    mqtt_connect_info.connected = false;
    pthread_cond_signal(&mqtt_cond);
}

static void onDisconnect(void *context, MQTTAsync_successData *response)
{
    printf("Successful disconnection\n");
}

static void onConnectFailure(void *context, MQTTAsync_failureData *response)
{
    printf("Connect failed, rc:%d ,msg:%s\n", response ? response->code : 0, response ? response->message : "NULL");
}

static void onSubscribe(void *context, MQTTAsync_successData *response)
{
    printf("Subscribe succeeded\n");
}

static void onSubscribeFailure(void *context, MQTTAsync_failureData *response)
{
    printf("Subscribe failed, rc %d\n", response ? response->code : 0);
}

static void onConnect(void *context, MQTTAsync_successData *response)
{
    char self_topic[128]           = TOPIC_COMMAND_SELF;
    MQTTAsync client               = (MQTTAsync)context;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    int rc;

    mqtt_connect_info.connected = true;
    printf("Successful connection\n");

    opts.onSuccess = onSubscribe;
    opts.onFailure = onSubscribeFailure;
    opts.context   = client;

    //从配置文件获取设备ID
    rc = get_string_value_form_config(CONFIG_KEY_SYSTEM, "dev_id", CONFIG_KEY_CONFIG_NAME, g_lst_var.dev_id);
    if(rc != 0)
    {
        printf("Failed to get dev id, return code %d\n", rc);
    }

    strncat(self_topic, g_lst_var.dev_id, DEV_ID_LEN);
    printf("self_topic:%s\n", self_topic);
    if((rc = MQTTAsync_subscribe(client, self_topic, MQTT_QOS, &opts)) != MQTTASYNC_SUCCESS)
    {
        printf("Failed to start subscribe self, return code %d\n", rc);
        exit(-1);
    }
    else
    {
        printf("Subscribe self success\n");
    }

    if((rc = MQTTAsync_subscribe(client, TOPIC_COMMAND_ALL, MQTT_QOS, &opts)) != MQTTASYNC_SUCCESS)
    {
        printf("Failed to start subscribe self, return code %d\n", rc);
        exit(-2);
    }
    else
    {
        printf("Subscribe all success\n");
    }
}

//接收到MQTT数据
static int mqtt_msg_recv(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
    printf("Message arrived\n");
    printf("topic: %s\n", topicName);

    deal_recv_command(message->payload, message->payloadlen);

    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topicName);
    return 1;
}

//连接MQTT服务器
static int mqtt_connect_server()
{
    int rc;
    MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession      = 1;
    conn_opts.onSuccess         = onConnect;
    conn_opts.onFailure         = onConnectFailure;
    conn_opts.context           = mqtt_connect_info.client;
    conn_opts.username          = mqtt_token_response.clientId;
    conn_opts.password          = mqtt_token_response.token;
    if((rc = MQTTAsync_connect(mqtt_connect_info.client, &conn_opts)) != MQTTASYNC_SUCCESS)
    {
        printf("Failed to start connect, return code %d\n", rc);
        return MQTTASYNC_FAILURE;
    }
    return MQTTASYNC_SUCCESS;
}

static void mqtt_disconnect_server()
{
    int rc;
    MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
    disc_opts.onSuccess                   = onDisconnect;
    if((rc = MQTTAsync_disconnect(mqtt_connect_info.client, &disc_opts)) != MQTTASYNC_SUCCESS)
    {
        printf("Failed to start disconnect, return code %d\n", rc);
    }
}

//MQTT服务线程
static void *mqtt_ps_thread(void *args)
{
    int rc;

    rc = get_string_value_form_config(CONFIG_KEY_MQTT_SERVER, "mqtt_addr", CONFIG_KEY_CONFIG_NAME, mqtt_connect_info.mqtt_addr);
    if(rc != 0)
    {
        printf("Failed to get mqtt addr, return code %d\n", rc);
    }

    while(1)
    {
        if(mqtt_connect_info.connected == false)
        {

            rc = get_token_from_shell(&mqtt_token_response);
            if(rc != 0)
            {
                printf("get token error:%d\n", rc);
            }
            else
            {
                MQTTAsync_create(&mqtt_connect_info.client, mqtt_connect_info.mqtt_addr, mqtt_token_response.clientId, 1, NULL);
                MQTTAsync_setCallbacks(mqtt_connect_info.client, NULL, connlost, mqtt_msg_recv, NULL);
                mqtt_connect_server();
            }
        }

        if(mqtt_connect_info.connected == true)
        {
            pthread_mutex_lock(&mqtt_mutex);
            pthread_cond_wait(&mqtt_cond, &mqtt_mutex);
            pthread_mutex_unlock(&mqtt_mutex);
        }
        else
        {
            printf("mqtt is reconnecting...\n");
            sleep(10); //10s重连
        }
    }

    //断开连接
    mqtt_disconnect_server();
    MQTTAsync_destroy(&mqtt_connect_info.client);

    return NULL;
}

//通过MQTT上传数据
void mqtt_upload_datas(uint8_t *data, int len, char *topic)
{
    if(mqtt_connect_info.connected == false)
    {
        printf("MQTT NOT READY\n");
        return;
    }
    int rc;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    MQTTAsync_message pubmsg       = MQTTAsync_message_initializer;

    opts.context = mqtt_connect_info.client;

    pubmsg.payload    = data;
    pubmsg.payloadlen = len;
    pubmsg.qos        = MQTT_QOS;
    pubmsg.retained   = 0;

    if((rc = MQTTAsync_sendMessage(mqtt_connect_info.client, topic, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
    {
        printf("Failed to start sendMessage, return code %d\n", rc);
        mqtt_connect_info.connected = false;
        pthread_cond_signal(&mqtt_cond);
    }
    else
    {
        // printf("send succuss\n");
    }
}

//MQTT服务初始化
int mqtt_service_init()
{
    pthread_t tid;
    pthread_create(&tid, NULL, mqtt_ps_thread, NULL);
    pthread_detach(tid);
    return 0;
}